package cn.pugle.base.thread;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 利用condition实现生产者消费者的"有界阻塞队列"
 *
 * @author tzp
 * @since 2018/2/8
 */
public class ConditionTest {
    public static void main(String[] args) {
        LimitBlockQueue<Object> blockQueue = new LimitBlockQueue<>(2);
        final AtomicInteger counter = new AtomicInteger(0);

        Runnable producer = () -> {
            try {
                while (!Thread.interrupted() && counter.get() <= 8) {
                    blockQueue.add(counter.incrementAndGet());
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        Runnable consumer = () -> {
            try {
                while (!Thread.interrupted()) {
                    blockQueue.remove();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        new Thread(producer).start();
        new Thread(producer).start();
        new Thread(consumer).start();


    }

    static class LimitBlockQueue<T> {
        private Object[] data;
        private int currentNum = 0, addPos = 0, removePos = 0;
        private Lock lock = new ReentrantLock();
        private Condition notEmpty = lock.newCondition();
        private Condition notFull = lock.newCondition();


        public LimitBlockQueue(int length) {
            this.data = new Object[length];
        }

        public void add(T ele) throws InterruptedException {
            lock.lock();
            try {
                while (currentNum == data.length) {//这个循环是保险起见吧
                    System.out.println(Thread.currentThread() + "= F");
                    notFull.await();
                }
                data[addPos] = ele;
                if (++addPos == data.length) {
                    addPos = 0;
                }
                currentNum++;
                notEmpty.signal();
                System.out.println(Thread.currentThread() + "+ " + ele);
            } finally {
                lock.unlock();
            }
        }

        public T remove() throws InterruptedException {
            lock.lock();
            try {
                while (currentNum == 0) {
                    System.out.println(Thread.currentThread() + "= E");
                    notEmpty.await();
                }
                T ele = (T) data[removePos];
                data[removePos] = null;
                if (++removePos == data.length) {
                    removePos = 0;
                }
                currentNum--;
                notFull.signal();
                System.out.println(Thread.currentThread() + "- " + ele);
                return ele;
            } finally {
                lock.unlock();
            }
        }
    }
}
